package org.jeecg.modules.fund.util;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.modules.fund.entity.FundLibrary;
import org.jeecg.modules.fund.entity.FundValue;
import org.jeecg.modules.fund.mapper.FundLibraryMapper;
import org.jeecg.modules.fund.mapper.FundValueMapper;
import org.jeecg.modules.fund.service.IFundLibraryService;
import org.jeecg.modules.fund.service.IFundValueService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

@Deprecated
@Slf4j
@Component
public class BatchDataProcessor {
    @Autowired
    private IFundValueService fundValueService;
    @Autowired
    private FundValueMapper fundValueMapper;
    @Autowired
    private IFundLibraryService fundLibraryService;
    @Autowired
    private FundLibraryMapper fundLibraryMapper;
    // 配置参数
    private static final int CORE_POOL_SIZE = 30; // 核心线程数（建议CPU核心数*2-5）
    private static final int MAX_POOL_SIZE = 100; // 最大线程数（网络IO密集型可设高）
    private static final int QUEUE_CAPACITY = 24000;     // 阻塞队列容量（防止OOM）
    private static final int BATCH_SIZE = 100;         // 批处理大小
    private final ExecutorService executor;

    public BatchDataProcessor() {
        // 创建定制线程池
        this.executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满时由主线程执行
        );
    }

    /**
     * 处理更新任务
     *
     * @param
     */
    public void processBatch() {
        log.info("开始测试时间：" + DateUtil.now());
        // 根据规则获取代码
        List<FundLibrary> categoryList = fundLibraryService.list(new QueryWrapper<FundLibrary>()
                .select("RIGHT(code,2) as 'code'")
                .groupBy("RIGHT(code,2)"));
        // 批次数组
        List<List<FundLibrary>> batchList = new ArrayList<>();
        for (FundLibrary category : categoryList){
            // 1.获取每个类别的数据
            List<FundLibrary> fundLibraryList = fundLibraryService.list(new QueryWrapper<FundLibrary>()
                    .ne("type", "货币型")
                    .eq("RIGHT(code,2)", category.getCode()));
            // 2.对每个类别数据再进行数据拆分
            // 总数量
            int size = fundLibraryList.size();
            // 遍历获取批次
            for (int i = 0; i < size; i += BATCH_SIZE) {
                // 获取每个结尾下标值
                int end = Math.min(i + BATCH_SIZE, size);
                // 数据提前
                List<FundLibrary> subList = fundLibraryList.subList(i, end);
                // 加到批次列表
                batchList.add(subList);
            }
        }
        log.info("总批次");

        // 分批处理数据
        List<CompletableFuture<Void>> futures = new ArrayList<>();
        // 遍历批次
        for(List<FundLibrary> item : batchList){
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                // 处理单个批次
                processSingleBatch(item);
            },executor);
            futures.add(future);
        }
        // 等待所有任务完成
        CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
        try {
            allFutures.get(); // 阻塞当前线程直到所有任务完成
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace(); // 异常情况处理
        } finally {
            log.info("结束测试时间：" + DateUtil.now());
            executor.shutdown(); // 关闭线程池
        }
    }

    /**
     * 批量处理单一批次数据
     *
     * @param fundLibraryList
     */
    private void processSingleBatch(List<FundLibrary> fundLibraryList) {
        // 获取表名
        String tableName = fundValueService.tableName(fundLibraryList.get(0).getCode());
        // 1.批量拉取远程数据
        List<FundValue> remoteData = fetchRemoteDataBatch(fundLibraryList);
        // 2.批量更新远程数据
        if(!remoteData.isEmpty()){
            // 日期
            String date = DateUtil.format(remoteData.get(0).getValDate(), "yyyy-MM-dd");
            // 更新子表，上面已经有限制如果是净值，不更新
            fundValueMapper.batchInsertOrUpdate(tableName, remoteData);
            // 获取line60
            List<FundValue> avgList = fundValueMapper.findValueAvgByDate(tableName, date);
            // 更新子表的line60
            fundValueMapper.batchInsertOrUpdate(tableName, avgList);
            // 更新主表的估值及line60
            fundLibraryMapper.updateEstimateLine60(avgList);
        }
    }

    /**
     * 批量获取数据
     *
     * @param fundLibraryList
     * @return
     */
    private List<FundValue> fetchRemoteDataBatch(List<FundLibrary> fundLibraryList) {
        List<FundValue> result = new ArrayList<>();
        for (FundLibrary library : fundLibraryList) {
            // 请求接口获取估值
            JSONObject object = FundDataUtil.getEstimateVal(library.getCode());
            // 如果没有估算数据，就跳过、QDII没有估值
            if (object != null && !"QDII".equals(library.getType())) {
                // ******* 构造子表实体 *********
                // 日期，注意存在空的情况
                String date = object.getString("gztime");
                // FundId
                String fundId = library.getId();
                // 子表ID
                String valId = fundValueService.createId(fundId, DateUtil.parseDate(date));
                //
                String tableName = fundValueService.tableName(library.getCode());
                // 查询净值
                FundValue value = fundValueMapper.findById(tableName, valId);
                // 空值或者估值，不更新
                if (date != null && (value == null || value.getIsEstimate() == 1)) {
                    // 实体构造
                    FundValue valueEntity = new FundValue();
                    // 设置主键
                    valueEntity.setId(valId);
                    // 净值日期
                    valueEntity.setValDate(DateUtil.parseDate(date));
                    // 所属基金ID
                    valueEntity.setFundId(fundId);
                    // 单位净值
                    valueEntity.setNowVal(object.getBigDecimal("gsz"));
                    // 日增长率
                    valueEntity.setValRate(object.getBigDecimal("gszzl"));
                    // 设置状态为估算状态
                    valueEntity.setIsEstimate(1);
                    // 设置创建时间
                    valueEntity.setCreateTime(new Date());
                    // 设置更新时间
                    valueEntity.setUpdateTime(new Date());
                    // 加入列表
                    result.add(valueEntity);
                }
            }
        }
        return result;
    }
}
